package com.zhaotianhao.springboot_demo_fire.task;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhaotianhao.springboot_demo_fire.config.MyThread;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Set;

/**
 * @Author Z-7
 * @Date 2022/3/22
 */
@Slf4j
@Component
public class FireTask {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    //@PostConstruct
    public void myThread(){
        threadPoolTaskExecutor.execute(()->{
            while (true){
                String name = Thread.currentThread().getName();
                log.info("线程"+ name +"正在执行");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    @Scheduled(cron = "0 0 * * * ?")
    public void tellTime(){
        log.info("当前时间：{}", DateUtil.now());
    }
    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private RedisTemplate redisTemplate;
    /**
     * 每分钟检测一次
     */
    @Scheduled(cron = "0 * * * * ?")
    public void messageScan(){
        Set keys = redisTemplate.keys("message:*");
        if(keys.size() == 0){
            log.info("暂时没有未消费的消息");
            return;
        }
        log.info("开始处理未消费的消息");
        for (Object key : keys) {
            Object o = redisTemplate.opsForValue().get(key);
            JSONObject jsonObject = JSON.parseObject(o.toString());
            if(System.currentTimeMillis() - jsonObject.getLong("create_time") > 5000){
                // 5秒后还没消费 重新发送
                kafkaTemplate.send("fire",jsonObject.getString("operation"),jsonObject.getString("message"));
                log.info("kafka重新发送");
            }
        }
    }
}
